package com.clei.Y2023.M09.D15;

import com.clei.utils.FileUtil;
import com.clei.utils.PrintUtil;

import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.stream.Collectors;

/**
 * 数据map
 * 之前有个架构是这样的，开启三个数据服务接收第三方数据，然后丢到RabbitMQ（没开启持久化）
 * 然后这三个数据服务又消费这个消息，丢到Redis里
 * 想给它优化一下，去掉RabbitMQ，要保证数据不丢的话还要考虑下SpringBoot的优雅停机相关
 * <p>
 * 放到map里，消费者每秒消费（批量处理），突然宕机也只会丢失1秒的数据量
 *
 * @author Y
 * @date 2023-09-15
 **/
public class DataMap {

    /**
     * 初始容量 参考qps
     */
    private final static int MAP_CAPACITY = 500;

    /**
     * 当前使用的map
     */
    private volatile Map<String, String> dataMap = new ConcurrentHashMap<>(MAP_CAPACITY);

    /**
     * 切换状态 true切换中 false未切换
     */
    private volatile boolean change = false;

    /**
     * lock
     */
    private final Object lock = new Object();

    public static void main(String[] args) throws Exception {

        // 数据容器
        DataMap dataMap = new DataMap();

        // 数据访问线程池
        ThreadPoolExecutor dataProducerExecutor = new ThreadPoolExecutor(
                4,
                9,
                30,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(100000),
                new ThreadPoolExecutor.CallerRunsPolicy());
        // 数据接收方
        ScheduledExecutorService dataConsumerExecutor = Executors.newSingleThreadScheduledExecutor();
        // 控制数据制造时间的线程
        Thread dataProducerTime = new Thread(() -> {
            try {
                // 10秒后停止
                TimeUnit.SECONDS.sleep(10);
                dataProducerExecutor.shutdownNow();
                dataConsumerExecutor.shutdownNow();
            } catch (Exception e) {
                PrintUtil.log("dataProducerTime sleep error ", e);
            }
        });
        // 放数据的线程
        Thread dataProducer = new Thread(() -> {
            Random random = new Random();
            try {
                for (int i = 0; !dataProducerExecutor.isTerminated(); i++) {
                    String key = String.valueOf(i);
                    dataProducerExecutor.execute(() -> {
                        dataMap.put(key, key);
                    });
                    TimeUnit.MILLISECONDS.sleep(random.nextInt(5));
                }
            } catch (Exception e) {
                PrintUtil.log("dataProducer error ", e);
            }
        });
        dataProducer.start();
        dataProducerTime.start();

        // 每秒消费一次
        dataConsumerExecutor.scheduleAtFixedRate(() -> {
            dataMap.consume(map -> {
                List<String> lineList = map.entrySet().stream()
                        .map(e -> e.getKey() + ' ' + e.getValue())
                        .collect(Collectors.toList());
                PrintUtil.log("消费数据数： {}", lineList.size());
                FileUtil.writeTxt("E:\\workTemp\\dataMap.txt", lineList, true);
            });
        }, 1, 1, TimeUnit.SECONDS);
    }

    /**
     * put
     *
     * @param key   key
     * @param value value
     */
    public void put(String key, String value) {
        if (change) {
            synchronized (lock) {
                PrintUtil.log("阻塞");
            }
        }
        dataMap.put(key, value);
    }

    /**
     * 消费数据
     *
     * @param consumer 消费者
     */
    public void consume(Consumer<Map<String, String>> consumer) {
        Map<String, String> tempMap = null;
        synchronized (lock) {
            // 锁后才修改change
            change = true;
            // 等待put结束
            try {
                TimeUnit.MILLISECONDS.sleep(5);
            } catch (Exception e) {
                PrintUtil.log("等待put结束sleep error ", e);
            }
            tempMap = dataMap;
            // 切换
            dataMap = new ConcurrentHashMap<>(MAP_CAPACITY);
        }
        change = false;
        consumer.accept(tempMap);
    }
}
